package com.fwj.storm.sentence;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;
import org.apache.storm.utils.Utils;

/**
 * @author 傅文江
 * <p>
 * 时间： 2017-11-07 15:03
 * <p>
 * 描述：
 * <p>
 * 版权： Copyright © 2017 新道科技股份有限公司
 **/
public class WordCountTopology {

    private static final String SENTENCE_SPOUT_ID = "sentence-spout";
    private static final String SPLIT_BOLT_ID = "split-bolt";
    private static final String COUNT_BOLT_ID = "count-bolt";
    private static final String REPORT_BOLT_ID = "report-bolt";
    private static final String TOPOLOGY_NAME = "word-count-topology";

    public static void main(String[] args) throws Exception {
        SentenceSpout spout = new SentenceSpout();
        SplitSentenceBolt splitBolt = new SplitSentenceBolt();
        WordCountBolt countBolt = new WordCountBolt();
        ReportBolt reportBolt = new ReportBolt();

        TopologyBuilder topologyBuilder = new TopologyBuilder();
        //设置SentenceSpout并发为2个线程
        topologyBuilder.setSpout(SENTENCE_SPOUT_ID, spout,2);
        //shuffleGrouping()方法 告诉Storm，要将类SentenceSpout发射的tuple随机均匀的分发给SplitSentenceBolt的实例
        //设置SplitSentenceBolt为2个Executor，4个task
        topologyBuilder.setBolt(SPLIT_BOLT_ID, splitBolt,2).setNumTasks(4).shuffleGrouping(SENTENCE_SPOUT_ID);
        //fieldsGrouping()方法保证所有的"word"字段值相同的tuple会被路由到同一个WordCountBolt实例中
        topologyBuilder.setBolt(COUNT_BOLT_ID, countBolt,4).fieldsGrouping(SPLIT_BOLT_ID, new Fields("word"));
        //globalGrouping()方法让WordCount发射的所有的tuple路由到唯一的Report任务中。
        topologyBuilder.setBolt(REPORT_BOLT_ID, reportBolt).globalGrouping(COUNT_BOLT_ID);

        Config config = new Config();
        //指派两个Workers进行运算（分配两个JVM java 虚拟机）
        config.setNumWorkers(2);
        if(args.length==0) {
            LocalCluster cluster = new LocalCluster();
            cluster.submitTopology(TOPOLOGY_NAME, config, topologyBuilder.createTopology());
            Utils.sleep(10000);
            cluster.killTopology(TOPOLOGY_NAME);
            cluster.shutdown();
        }else{
            StormSubmitter.submitTopology(args[0],config,topologyBuilder.createTopology());
        }
    }


}
